package org.ala.tcp_feign_client.spring.feign;

import java.util.ArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

import org.ala.tcp_feign_client.ITcpFeignCallback;
import org.ala.tcp_feign_client.ITcpFeignClient;
import org.ala.tcp_feign_client.context.TcpFeignClientContext;
import org.ala.tcp_feign_client.spring.feign.clients.ITcpFeignClientHolder;
import org.ala.tcp_feign_client.spring.feign.clients.impl.EurekaTcpFeignClientHolder;
import org.ala.tcp_feign_client.spring.feign.clients.impl.HostsTcpFeignClientHolder;
import org.ala.tcp_feign_client.spring.feign.id_generator.DefaultTcpFeignClientMsgIdGenerator;
import org.ala.tcp_feign_client.spring.feign.id_generator.ITcpFeignClientMsgIdGenerator;
import org.ala.tcp_feign_client.spring.feign.results.DefaultTcpFeignClientResultsMapper;
import org.ala.tcp_feign_client.spring.feign.results.IBizRespResultCallback;
import org.ala.tcp_feign_client.spring.feign.results.ITcpFeignClientResultsMapper;
import org.ala.tcp_feign_client.spring.feign.results.impl.BizRespResultAsyncCallback;
import org.ala.tcp_feign_client.spring.feign.results.impl.BizRespResultSAsyncCallback;
import org.ala.tcp_feign_client.spring.feign.results.impl.BizRespResultSyncCallback;
import org.ala.tcp_feign_client.spring.feign.selector.DefaultTcpFeignClientSelector;
import org.ala.tcp_feign_client.spring.feign.selector.ITcpFeignClientSelector;
import org.ala.tcp_feign_common.config.TcpFeignConfigure;
import org.ala.tcp_feign_common.exceptions.ActiveException;
import org.ala.tcp_feign_common.exceptions.LoginException;
import org.ala.tcp_feign_common.exceptions.RemoteInvokeException;
import org.ala.tcp_feign_common.exceptions.SerializableException;
import org.ala.tcp_feign_common.logs.LoggerFactory;
import org.ala.tcp_feign_common.msg.biz.BizByteMessage;
import org.ala.tcp_feign_common.msg.biz.body.BizBody;
import org.ala.tcp_feign_common.msg.biz_resp.BizRespByteMessage;
import org.slf4j.Logger;
import org.springframework.util.StringUtils;


/**
 * tcp-feign
 * <p>	负责持有链接，消息往来，接收resp
 *
 * @author 骆毅(ala)
 * @date 2024年4月22日
 */
public class TcpFeign {
	
	
	static Logger log = LoggerFactory.getLogger("tcp-feign", "client");
	

	/**
	 * 相关配置
	 */
	protected TcpFeignConfigure configure;
	
	/**
	 * 远程服务名称
	 */
	protected String remoteName;
	/**
	 * 远程服务host，port
	 */
	protected String hosts;
	/**
	 * 根url
	 */
	protected String path;
	/**
	 * 是否专线
	 */
	protected boolean dedicated;
	/**
	 * 被代理的接口类型
	 */
	protected Class<?> type;
	
	
	/**
	 * 后缀
	 */
	protected String suffix = TcpFeignClientContext.SUFFIX_COMMON;
	/**
	 * client holder
	 */
	protected ITcpFeignClientHolder clientHolder;
	/**
	 * client selector
	 */
	protected ITcpFeignClientSelector clientSelector;
	/**
	 * client result mapper
	 */
	protected ITcpFeignClientResultsMapper clientResultMapper;
	/**
	 * client id generator
	 */
	protected ITcpFeignClientMsgIdGenerator clientIdGenerator;
	
	
	/**
	 * 初始化
	 */
	public void init() {
		if (dedicated) {
			suffix = type.getSimpleName();
		}
		
		//	初始化链接selector
		clientSelector = new DefaultTcpFeignClientSelector();
		//	初始化resp
		clientResultMapper = new DefaultTcpFeignClientResultsMapper(configure.getBizTimeout(), dedicated);
		//	id生成器
		clientIdGenerator = new DefaultTcpFeignClientMsgIdGenerator();
		
		//	初始化链接holder
		//	优先使用name属性
		if (!StringUtils.isEmpty(remoteName)) {
			clientHolder = new EurekaTcpFeignClientHolder(suffix, new BizRespConsumer(clientResultMapper));
			clientHolder.initClients(remoteName);
		}
		//	否则使用hosts配置
		else {
			//	收到返回值回调（跑在ClientBizResp线程池里，在那个线程里通知Future给值）
			clientHolder = new HostsTcpFeignClientHolder(suffix, new BizRespConsumer(clientResultMapper));
			clientHolder.initClients(hosts);
		}
		
		

	}
	
	
	/**
	 * 构建BizByteMessage
	 * @throws SerializableException 
	 */
	protected BizByteMessage buildMsg(long msgId, String url, Object[] args) throws SerializableException {
		BizBody body = BizBody.build(url, args);
		byte[] bytes = TcpFeignClientContext.instance().getSerializabler().serialize(body);
		
		BizByteMessage msg = BizByteMessage.build(msgId, bytes);
		return msg;
	}
	
	
	/**
	 * 执行方法
	 * 
	 * @param	asyncType: 0:同步 | 1:半异步 | 2:异步
	 * 
	 * <p>	3种实现方式：
	 * <p>		1: 接口直接返回结果：同步
	 * <p>		2: 接口的返回结果是Future类型：半异步
	 * <p>		3: 接口的最后一个参数是ITcpFeignCallback类型：异步
	 */
	public Object invoke(String url, int asyncType, Object[] args, Class<?> returnType) throws RemoteInvokeException {
		//	本次要通讯的消息id
		long msgId = clientIdGenerator.generatorId();
		//	选择client
		ITcpFeignClient client = clientSelector.select(clientHolder.all());
		if (client == null) {
			log.error("[TcpFeign invoke] 当前没有有效连接 url:" + url + " bean:" + type.getName());
			throw new RuntimeException("无法找到有效的client端链接");
		}
		
		//	根据异步方式不同选择不同的处理方案
		switch (asyncType) {
			//	同步
			case TcpFeignInvocationHandler.SYNC: {
				return invokeSync(msgId, url, args, client, returnType);
			}
			//	半异步
			case TcpFeignInvocationHandler.SASYNC: {
				return invokeSAsync(msgId, url, args, client, returnType);
			}
			//	异步
			case TcpFeignInvocationHandler.ASYNC: {
				invokeAsync(msgId, url, args, client, returnType);
				return null;
			}
			//	默认同步
			default: {
				return invokeSync(msgId, url, args, client, returnType);
			}
		}
	}
	
	
	/**
	 * 同步处理方式
	 * <p>	同步等待返回结果
	 */
	protected Object invokeSync(long msgId, String url, Object[] args, ITcpFeignClient client, Class<?> returnType) {
		//	构建msg
		try {
			BizByteMessage msg = buildMsg(msgId, url, args);
			
			//	用msgId绑定result，等待回调
			//	msgId会贯穿：Biz消息 -> BizACK，BizResp -> BizRespACK整个流程
			BizRespResultSyncCallback callback = new BizRespResultSyncCallback(msgId, configure.getBizTimeout(), clientResultMapper);
			clientResultMapper.bind(msg.getMsgId(), callback);
			callback.setBindTimestamp(System.currentTimeMillis());
			
			//	发送消息
			client.send(msg);
			
			//	同步等待返回值
			return callback.getFuture().get(configure.getBizTimeout(), TimeUnit.MILLISECONDS);
		} catch (SerializableException | ActiveException | LoginException | InterruptedException | ExecutionException | TimeoutException e) {
			clientResultMapper.unbind(msgId);
			throw new RuntimeException(e);
		}
	}
	/**
	 * 半异步处理方式
	 * <p>	返回结果是Future类型
	 */
	protected Object invokeSAsync(long msgId, String url, Object[] args, ITcpFeignClient client, Class<?> returnType) {
		//	构建msg
		try {
			BizByteMessage msg = buildMsg(msgId, url, args);
			
			//	用msgId绑定result，等待回调
			//	msgId会贯穿：Biz消息 -> BizACK，BizResp -> BizRespACK整个流程
			BizRespResultSAsyncCallback callback = new BizRespResultSAsyncCallback(msgId, configure.getBizTimeout(), clientResultMapper);
			clientResultMapper.bind(msg.getMsgId(), callback);
			callback.setBindTimestamp(System.currentTimeMillis());
			
			//	发送消息
			client.send(msg);
			
			//	同步等待返回值
			return callback.getFuture();
		} catch (SerializableException | ActiveException | LoginException e) {
			clientResultMapper.unbind(msgId);
			throw new RuntimeException(e);
		}
	}
	/**
	 * 异步处理方式
	 */
	@SuppressWarnings("rawtypes")
	protected void invokeAsync(long msgId, String url, Object[] args, ITcpFeignClient client, Class<?> returnType) {
		//	构建msg
		try {
			//	最后一个参数是ITcpFeignCallback类型
			ITcpFeignCallback tcb = (ITcpFeignCallback) args[args.length - 1];
			Object[] newArgs = new Object[args.length-1];
			if (newArgs.length > 0) { System.arraycopy(args, 0, newArgs, 0, newArgs.length); }
			BizByteMessage msg = buildMsg(msgId, url, newArgs);
			
			//	用msgId绑定result，等待回调
			//	msgId会贯穿：Biz消息 -> BizACK，BizResp -> BizRespACK整个流程
			BizRespResultAsyncCallback callback = new BizRespResultAsyncCallback(msgId, configure.getBizTimeout(), clientResultMapper);
			callback.setCallback(tcb);
			clientResultMapper.bind(msg.getMsgId(), callback);
			callback.setBindTimestamp(System.currentTimeMillis());
			
			//	发送消息
			client.send(msg);
		} catch (SerializableException | ActiveException | LoginException e) {
			clientResultMapper.unbind(msgId);
			throw new RuntimeException(e);
		}
	}
	
	
	/**
	 * 登录成功回调
	 */
	public static class LoginSuccessConsumer implements Consumer<ITcpFeignClient> {
		protected int i;
		protected ArrayList<ITcpFeignClient> clients;
		public LoginSuccessConsumer(int i, ArrayList<ITcpFeignClient> clients) {
			super();
			this.i = i;
			this.clients = clients;
		}
		@Override
		public void accept(ITcpFeignClient t) {
			clients.set(i, t);
		}
	}
	/**
	 * resp消息回调
	 */
	public static class BizRespConsumer implements Consumer<BizRespByteMessage> {
		ITcpFeignClientResultsMapper clientResultMapper;
		
		public BizRespConsumer(ITcpFeignClientResultsMapper clientResultMapper) {
			super();
			this.clientResultMapper = clientResultMapper;
		}

		@Override
		public void accept(BizRespByteMessage t) {
			IBizRespResultCallback callback = clientResultMapper.getAndRemove(t.getMsgId());
			if (callback != null) {
				try {
					callback.doCallback(t);
				} catch (SerializableException e) {
					throw new RuntimeException("消息反序列化错误", e);
				}
			}
		}
	}
	
	
	public TcpFeignConfigure getConfigure() {
		return configure;
	}
	public void setConfigure(TcpFeignConfigure configure) {
		this.configure = configure;
	}
	public String getRemoteName() {
		return remoteName;
	}
	public void setRemoteName(String remoteName) {
		this.remoteName = remoteName;
	}
	public String getHosts() {
		return hosts;
	}
	public void setHosts(String hosts) {
		this.hosts = hosts;
	}
	public String getPath() {
		return path;
	}
	public void setPath(String path) {
		this.path = path;
	}
	public boolean isDedicated() {
		return dedicated;
	}
	public void setDedicated(boolean dedicated) {
		this.dedicated = dedicated;
	}
	public Class<?> getType() {
		return type;
	}
	public void setType(Class<?> type) {
		this.type = type;
	}

}
